package com.lvmama.rhino.streaming.nginx

import com.lvmama.rhino.common.entity.{JobConfig, KafkaTopic}
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.kafka.{KafkaDStreamSource, KafkaPayloadStringCodec}
import com.lvmama.rhino.common.utils.spark.SparkStreamingApplication
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.StreamingContext

import scala.util.parsing.json.JSONObject

/**
  * Created by yuanxiaofeng on 2016/6/23.
  */
class ManGodStatStreaming(config: JobConfig) extends SparkStreamingApplication{
  private val inputTopic = KafkaTopic("manGodTopic")
  private val outputTopic = KafkaTopic("manGodResultTopic")

  override var appName: String = "ManGodStatStreaming"

  override def streamingBatchDuration: Long = 60L

  val streamingCheckpointDir: Map[String, String] = config.streamingCheckpointDir

  override var sparkConfig: Map[String, String] = config.spark

  def execute(): Unit = {
    sparkConfig += ("spark.app.name" -> appName)
//    sparkConfig += ("spark.master" -> "local[4]")
    sparkConfig += ("spark.kryo.registrator" -> "com.lvmama.rhino.common.utils.serialization.ManGodStreamingRegistrator")
    withSparkStreamingContext{ (sc, ssc) =>
      val input = KafkaDStreamSource(config.sourceKafka).createSource(ssc, inputTopic.name)
      val stringCodec = sc.broadcast(KafkaPayloadStringCodec())
      val lines = input.flatMap(stringCodec.value.decodeValue(_))
      val result = mangodStat(ssc, lines)
      val output = result.map(r => JSONObject.apply(r.toMap).toString()).map(stringCodec.value.encodeValue(_))
      import com.lvmama.rhino.common.utils.kafka.KafkaDStreamSink._
      output.sendToKafka(config.sinkKafka, outputTopic.name)
    }
  }

  private def mangodStat(ssc: StreamingContext, lines: DStream[String]): DStream[Array[(String, Int)]] = {
    val sc = ssc.sparkContext
    val contrastMap = Map[String, String]("hm01" -> "hbActivityPage", "hm02" -> "barragePage", "hm03" -> "myHb",
      "hm01_c01" -> "obtainAuthcode", "hm01_c02" -> "getHb", "hm02_c08" -> "sendTaunt")
    val contrastMapVar = sc.broadcast(contrastMap)

    val events = lines.repartition(4).map(line => Utils.initNginx(line))

    val forwardEvents = events.filter(_.logType == "forward").map(ng => (ng.pageCode, 1)).reduceByKey(_ + _)
    val clickEvents = events.filter(_.logType == "click").map(ng => (ng.pageCode + "_" + ng.eventId, 1)).reduceByKey(_ + _)

    forwardEvents.union(clickEvents).repartition(1).filter(r => r._1 == "hm01" || r._1 == "hm02" || r._1 == "hm03" || r._1 == "hm01_c01" || r._1 == "hm01_c02" || r._1 == "hm02_c08")
      .map{case (k, v) => {
      val map = contrastMapVar.value
      map(k) -> v
    }}.glom()
  }
}

object ManGodStatStreaming{
  def apply(): Unit = {
    val config = JobConfig()
    new ManGodStatStreaming(config).execute()
  }
}
